package com.dy.realtime.dwd;

import com.dy.realtime.common.base.BaseSqlApp;
import com.dy.realtime.common.constant.Constant;
import com.dy.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @projectName: MedicalTreatment
 * @package: com.dy.realtime.dwd
 * @className: DwdTradePrescriptionPaySuc
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/11 20:18
 * @version: 1.0
 * 交易域处方开单支付成功事务事实表
 */

public class DwdTradePrescriptionPaySuc extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradePrescriptionPaySuc().start(Constant.TOPIC_DWD_TRADE_PRESCRIPTION_PAY_SUC, Constant.TOPIC_DWD_TRADE_PRESCRIPTION_PAY_SUC, 4, 10014);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        readOdsDb(tableEnv, groupId);
        Table table = tableEnv.sqlQuery("select `data`['id']              as id,\n" +
                "       `data`['create_time']     as create_time,\n" +
                "       `data`['instruction']     as prescription_instruction,\n" +
                "       `data`['status']          as status,\n" +
                "       `data`['total_amount']    as total_amount,\n" +
                "       `data`['consultation_id'] as consultation_id,\n" +
                "       `data`['doctor_id']       as doctor_id,\n" +
                "       `data`['patient_id']      as patient_id,\n" +
                "       row_time\n" +
                "from  topic_db_medical\n" +
                "where `table` = 'prescription'\n" +
                "  and `type` = 'update' and `data`['status'] = '203';");
//        table.execute().print();
        tableEnv.createTemporaryView("prescription", table);



        Table table1 = tableEnv.sqlQuery("select `data`['id']              as id,\n" +
                "       `data`['create_time']     as create_time,\n" +
                "       `data`['count']           as `count`,\n" +
                "       `data`['instruction']     as prescription_detail_instruction,\n" +
                "       `data`['medicine_id']     as medicine_id,\n" +
                "       `data`['prescription_id'] as prescription_id,\n" +
                "       row_time\n" +
                "from topic_db_medical\n" +
                "where `table` = 'prescription_detail'\n" +
                "  and `type` = 'insert'");
//        table1.execute().print();
        tableEnv.createTemporaryView("prescription_detail", table1);



        Table result = tableEnv.sqlQuery("select p.id,\n" +
                "       p.create_time,\n" +
                "       prescription_instruction,\n" +
                "       status,\n" +
                "       total_amount,\n" +
                "       consultation_id,\n" +
                "       `count`,\n" +
                "       prescription_detail_instruction,\n" +
                "       prescription_id,\n" +
                "       doctor_id,\n" +
                "       patient_id,\n" +
                "       medicine_id\n" +
                "from  prescription p ,prescription_detail pd\n" +
                "where p.id = pd.prescription_id AND p.row_time BETWEEN pd.row_time - INTERVAL '15' MINUTE  AND pd.row_time + INTERVAL '5' MINUTE ");
//        result.execute().print();

        tableEnv.executeSql("create table "+Constant.TOPIC_DWD_TRADE_PRESCRIPTION_PAY_SUC+"\n" +
                "(\n" +
                "    id                              string,\n" +
                "    create_time                     string,\n" +
                "    prescription_instruction        string,\n" +
                "    status                          string,\n" +
                "    total_amount                    string,\n" +
                "    consultation_id                 string,\n" +
                "    `count`                         string,\n" +
                "    prescription_detail_instruction string,\n" +
                "    prescription_id string,\n" +
                "    doctor_id                       string,\n" +
                "    patient_id                      string,\n" +
                "    medicine_id                     string" +
                ")"+ SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_PRESCRIPTION_PAY_SUC));
        result.executeInsert(Constant.TOPIC_DWD_TRADE_PRESCRIPTION_PAY_SUC);

    }
}
